/**
 * 
 */
package edu.sjtu.seda.eventqueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;

import edu.sjtu.seda.event.Event;
import edu.sjtu.seda.log.LogService;
import edu.sjtu.seda.stage.Stage;
import edu.sjtu.seda.stage.StageWrapper;
import edu.sjtu.seda.threadpool.StageRunner;
import edu.sjtu.seda.threadpool.StageRunnerImpl;

/**
 * @author john.zhu
 * 
 */
public class EventQueueImpl implements EventQueue {
	private static LogService log = new LogService(EventQueueImpl.class);
	private BlockingQueue<Event> eventQueue;

	private StageWrapper wrapper;

	private int defaultSize = 10;

	private boolean enableBatch;
	private int batchSize = 10;

	public EventQueueImpl(StageWrapper wrapper, int size) {
		this(wrapper, size, false, 0);
	}

	/**
	 * event queue implementation. Receive events, wrap as stagerunner and send
	 * to threadpool finally send to next stage
	 * 
	 * @param wrapper
	 * @param size
	 */
	public EventQueueImpl(StageWrapper wrapper, int size, boolean enableBatch,
			int batchSize) {
		this.wrapper = wrapper;
		this.defaultSize = size;
		this.enableBatch = enableBatch;
		this.batchSize = batchSize;
		eventQueue = new LinkedBlockingQueue<Event>(defaultSize);
	}

	public void pushEvent(Event event) {
		if (this.eventQueue.offer(event)) {
			log.info("event " + event.getUID() + " has been pushed to queue "
					+ wrapper.getStage().getName());
		} else {
			/*
			 * reject events for the queue is full
			 */
			log.error("Queue " + wrapper.getStage().getName()
					+ " is full and event " + event.getUID()
					+ " will be rejected.");
		}
		synchronized (this) {
			notifyAll();
		}
	}

	public Event popEvent() {
		Event event = this.eventQueue.poll();
		log.info("event " + event.getUID() + " has been poped from queue "
				+ wrapper.getStage().getName());
		return event;
	}

	public Event element() {
		return this.eventQueue.peek();
	}

	public int size() {
		return this.eventQueue.size();
	}

	@Override
	public void clear() {
		this.eventQueue.clear();
	}

	@Override
	public void run() {
		Collection<StageRunner> batchList = new ArrayList<StageRunner>();

		while (true) {
			synchronized (this) {
				while (size() != 0) {
					try {
						Event event = element();
						if (event != null) {
							StageRunner runner = new StageRunnerImpl(event,
									wrapper.getEventHandler(), wrapper);
							if (enableBatch) {
								batchList.add(runner);
								if (batchList.size() >= batchSize) {
									for (StageRunner run : batchList) {
										wrapper.getThreadPool().execute(run)
												.get();
										Stage st = run.getEvent()
												.nextLocation();
										if (st != null) {
											st.getStageWrapper()
													.getEventQueue()
													.pushEvent(run.getEvent());
										} else {
											log.info("event "
													+ run.getEvent().getUID()
													+ " ended!");
										}
									}
									batchList = new ArrayList<StageRunner>();
								}
								popEvent();
							} else {
								wrapper.getThreadPool().execute(runner).get();
								popEvent();
								Stage st = event.nextLocation();
								if (st != null) {
									st.getStageWrapper().getEventQueue()
											.pushEvent(event);
								} else {
									log.info("event " + event.getUID()
											+ " ended!");
								}
							}
						} else {
							log.error("unknown error!");
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (ExecutionException e) {
						e.printStackTrace();
					}
				}
				try {
					wait();
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
			}
		}
	}

}
